package org.raftkv.raft.core.node;

import org.raftkv.commons.selector.Channel;
import org.raftkv.commons.selector.SelectionKey;
import org.raftkv.commons.selector.Selector;
import org.raftkv.raft.core.Raft;
import org.raftkv.raft.core.node.enums.SnapshotStatus;
import org.raftkv.raft.core.node.exceptions.StoppedException;
import org.raftkv.raft.core.rawnode.RawNode;
import org.raftkv.raft.core.status.Status;
import org.raftkv.raft.pb.ConfState;
import org.raftkv.raft.pb.Entry;
import org.raftkv.raft.pb.IConfChange;
import org.raftkv.raft.pb.Message;
import org.raftkv.raft.pb.enums.MessageType;
import org.raftkv.raft.pb.impl.ConfChangeV2;
import org.raftkv.raft.tracker.Progress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

import static org.raftkv.commons.selector.SelectionKey.*;

public class DefaultNode implements Node {

    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultNode.class);

    private Channel<Message> propc;

    private Channel<Message> recvc;

    private Channel<ConfChangeV2> confc;

    private Channel<ConfState> confstatec;

    private Channel<Ready> readyc;

    private Channel<Object> advancec;

    private Channel<Object> tickc;

    private Channel<Object> done;

    private Channel<Object> stop;

    private Channel<Channel<Status>> status;

    private RawNode rn;

    private DefaultNode() {

    }

    public static DefaultNode newNode(RawNode rn) {
        DefaultNode node = new DefaultNode();
        node.propc = new Channel<>();
        node.recvc = new Channel<>();
        node.confc = new Channel<>();
        node.confstatec = new Channel<>();
        node.readyc = new Channel<>();
        node.advancec = new Channel<>();

        node.tickc = new Channel<>(128);
        node.done = new Channel<>();
        node.stop = new Channel<>();
        node.status = new Channel<>();

        node.rn = rn;

        return node;
    }

    @Override
    public void stop() {
        SelectionKey<?> key = Selector.open()
                .register(this.stop, write(null))
                .register(this.done, read())
                .select();
        if (key.channel() == done) {
            return;
        }
        done.read();
    }

    public void run() {
        Channel<Message> propc = null;
        Channel<Ready> readyc = null;
        Channel<Object> advancec = null;
        Ready rd = new Ready();

        Raft r = this.rn.getRaft();

        long lead = Raft.NONE;

        while (true) {
            if (advancec != null) {
                readyc = null;
            } else if (this.rn.hasReady()) {
                rd = this.rn.readyWithoutAccept();
                readyc = this.readyc;
            }

            if (lead != r.getLead()) {
                if (r.hasLeader()) {
                    if (lead == Raft.NONE) {
                        LOGGER.info("raft.node: {} elected leader {} at term {}", r.getId(), r.getLead(), r.getTerm());
                    } else {
                        LOGGER.info("raft.node: {} changed leader from {} to {} at term {}", r.getId(), lead, r.getLead(), r.getTerm());
                    }
                    propc = this.propc;
                } else {
                    LOGGER.info("raft.node: {} lost leader {} at term {}", r.getId(), lead, r.getTerm());
                    propc = null;
                }
                lead = r.getLead();
            }


            SelectionKey<?> key = Selector.open()
                    .register(propc, read())
                    .register(recvc, read())
                    .register(confc, read())
                    .register(tickc, read())
                    .register(readyc, write(rd))
                    .register(advancec, read())
                    .register(status, read())
                    .register(stop, read())
                    .select();
            if (key.channel() == propc) {
                Message m = key.data(Message.class);
                m.setFrom(r.getId());
                r.step(m);
            } else if (key.channel() == recvc) {
                Message m = key.data(Message.class);
                Progress pr = r.getPrs().getProgress().getProgressMap().get(m.getFrom());
                if (pr != null || !Message.isResponseMsg(m.getType())) {
                    r.step(m);
                }
            } else if (key.channel() == confc) {
                ConfChangeV2 cc = key.data(ConfChangeV2.class);
                boolean okBefore = r.getPrs().getProgress().getProgressMap().containsKey(r.getId());
                ConfState cs = r.applyConfChange(cc);
                boolean okAfter = r.getPrs().getProgress().getProgressMap().containsKey(r.getId());

                if (okBefore && !okAfter) {
                    boolean found = false;
                    List<List<Long>> temp = new ArrayList<>();
                    temp.add(cs.getVoters());
                    temp.add(cs.getVotersOutgoing());

                    outer:
                    for (List<Long> sl : temp) {
                        for (Long id : sl) {
                            if (id == r.getId()) {
                                found = true;
                                break outer;
                            }
                        }
                    }

                    if (!found) {
                        propc = null;
                    }

                    Selector.open()
                            .register(confstatec, write(cs))
                            .register(done, read())
                            .select();
                }
            } else if (key.channel() == tickc) {
                rn.tick();
            } else if (key.channel() == readyc) {
                rn.acceptReady(rd);
                advancec = this.advancec;
            } else if (key.channel() == advancec) {
                rn.advance(rd);
                rd = new Ready();
                advancec = null;
            } else if (key.channel() == status) {
                Channel<Status> channel = ((Channel<Status>) key.data());
                Status st = Status.getStatus(r);
                channel.write(st);
            } else if (key.channel() == stop) {
                done.close();
                return;
            } else {
                String errMsg = "unknown channel";
                LOGGER.error(errMsg);
                throw new RuntimeException(errMsg);
            }
        }
    }

    @Override
    public void tick() {
        SelectionKey<?> key = Selector.open()
                .register(tickc, write(new Object()))
                .register(done, read())
                .fallback(fallback())
                .select();

        if (key.type() == FALLBACK) {
            LOGGER.warn("A tick missed to fire. Node blocks too long!");
        }
    }

    @Override
    public void campaign() {
        stepInternal(Message.builder()
                .type(MessageType.MsgHup)
                .build());
    }

    @Override
    public void propose(byte[] data) {
        stepInternal(Message.builder()
                .type(MessageType.MsgProp)
                .entries(List.of(new Entry(data)))
                .build());
    }

    @Override
    public void step(Message m) {
        if (Message.isLocalMsg(m.getType())) {
            return;
        }
        stepInternal(m);
    }

    @Override
    public void proposeConfChange(IConfChange cc) {
        Message msg = IConfChange.confChangeToMsg(cc);

        step(msg);
    }

    private void stepInternal(Message m) {
        Channel<Message> c = recvc;
        if (m.getType() == MessageType.MsgProp) {
            c = propc;
        }

        SelectionKey<?> key = Selector.open()
                .register(c, write(m))
                .register(done, read())
                .select();
        if (key.channel() == done) {
            throw new StoppedException();
        }
    }

    @Override
    public Channel<Ready> ready() {
        return readyc;
    }

    @Override
    public void advance() {
        Selector.open()
                .register(advancec, write(new Object()))
                .register(done, read())
                .select();
    }

    @Override
    public ConfState applyConfChange(IConfChange cc) {
        ConfState cs = new ConfState();
        Selector.open()
                .register(confc, write(cc.asV2()))
                .register(done, read())
                .select();

        SelectionKey<?> key = Selector.open()
                .register(confstatec, read())
                .register(done, read())
                .select();

        if (key.channel() == confstatec) {
            cs = key.data(ConfState.class);
        }
        return cs;
    }

    @Override
    public Status status() {
        Channel<Status> chan = new Channel<>();
        SelectionKey<?> key = Selector.open()
                .register(status, write(chan))
                .register(done, read())
                .select();
        if (key.channel() == status) {
            return chan.read();
        } else {
            return new Status();
        }
    }

    @Override
    public void reportUnreachable(long id) {
        Message m = new Message.MessageBuilder()
                .type(MessageType.MsgUnreachable)
                .from(id)
                .build();
        Selector.open()
                .register(recvc, write(m))
                .register(done, read())
                .select();
    }

    @Override
    public void reportSnapshot(long id, SnapshotStatus status) {
        boolean rej = status == SnapshotStatus.SnapshotFailure;
        Message m = new Message.MessageBuilder()
                .type(MessageType.MsgSnapStatus)
                .from(id)
                .reject(rej)
                .build();
        Selector.open()
                .register(recvc, write(m))
                .register(done, read())
                .select();
    }

    @Override
    public void transferLeaderShip(long lead, long transferee) {
        Message m = new Message.MessageBuilder()
                .type(MessageType.MsgTransferLeader)
                .from(transferee)
                .to(lead)
                .build();
        Selector.open()
                .register(recvc, write(m))
                .register(done, read())
                .select();
    }

    @Override
    public void readIndex(byte[] rctx) {
        Message m = new Message.MessageBuilder()
                .type(MessageType.MsgReadIndex)
                .entries(List.of(new Entry(rctx)))
                .build();
        stepInternal(m);
    }

}
